package com.raven.play.stream.test.scaladsl.tutorial

import akka.actor.ActorSystem
import akka.stream.Materializer
import akka.stream.scaladsl.{Flow, Keep, Sink, Source}

import scala.util.{Failure, Success}

object $06MaterializingStreams extends App {

  implicit val system = ActorSystem("MaterializingStreams")
  implicit val materializer = Materializer.createMaterializer(system)

  import system.dispatcher

  val simpleGraph = Source(1 to 10).to(Sink.foreach(println))
  // val simpleMaterializedValue = simpleGraph.run()
  // mat is NotUsed

  //  val source = Source(1 to 10)
  //  val sink = Sink.reduce[Int]((a, b) => a + b)
  //  val sumFuture = source.runWith(sink)
  //  // mat is Int
  //  sumFuture.onComplete {
  //    case Success(value) => println(s"The sum of all elements is $value")
  //    case Failure(ex) => println(s"The sum of the elements could not be computed $ex")
  //  }

  // choosing materialized values
  val simpleSource = Source(1 to 10)
  val simpleFlow = Flow[Int].map(x => x + 1)
  val simpleSink = Sink.foreach[Int](println)
  //  选择一个需要保留并输出的数据
  //  val graph = simpleSource.viaMat(simpleFlow)((source, flow) => source)
  //  val graph = simpleSource.viaMat(simpleFlow)((source, flow) => flow)
  //  val graph = simpleSource.viaMat(simpleFlow)(Keep.both)
  val graph = simpleSource.viaMat(simpleFlow)(Keep.right).toMat(simpleSink)(Keep.right)
  graph.run().onComplete {
    case Success(_) => println("Stream processing finished.")
    case Failure(exception) => println(s"Stream processing failed with:$exception")
  }

  // 可以看到 runWith 方法就是使用 toMat 方法实现的
  //  def runWith[Mat1, Mat2](source: Graph[SourceShape[In], Mat1], sink: Graph[SinkShape[Out], Mat2])(
  //    implicit materializer: Materializer): (Mat1, Mat2) =
  //    Source.fromGraph(source).via(this).toMat(sink)(Keep.both).run()

  // sugars
  Source(1 to 10).runWith(Sink.reduce((a: Int, b: Int) => a + b)) //source.to(Sink.reduce)(Keep.right)
  Source(1 to 10).runReduce((a: Int, b: Int) => a + b)

  // backwards
  Sink.foreach[Int](println).runWith(Source.single(42)) // source(..).to(sink...).run()
  // both ways
  Flow[Int].map(x => 2 * x).runWith(simpleSource, simpleSink)

  /**
   * - return the last element out of a source (use Sink.last)
   * - compute the total word count out of a stream of sentences
   *    - map, fold, reduce
   */
  val fl = Source(1 to 10).toMat(Sink.last)(Keep.right).run()
  val f2 = Source(1 to 10).runWith(Sink.last)
  val sentenceSource = Source(List(
    "Akka is awesome",
    "I lovestreams",
    "Materialized values are killing me"
  ))
  val wordCountSink = Sink.fold[Int, String](0)((currentWords, newSentence) => currentWords + newSentence.split(" ").length)
  val g1 = sentenceSource.toMat(wordCountSink)(Keep.right).run()
  val g2 = sentenceSource.runWith(wordCountSink)
  val g3 = sentenceSource.runFold(0)((currentWords, newSentence) => currentWords + newSentence.split(" ").length)
  val wordCountFlow = Flow[String].fold[Int](0)((currentWords, newSentence) => currentWords + newSentence.split(" ").length)
  val g4 = sentenceSource.via(wordCountFlow).toMat(Sink.head)(Keep.right).run()
  val g5 = sentenceSource.viaMat(wordCountFlow)(Keep.left).toMat(Sink.head)(Keep.right).run()
  val g6 = sentenceSource.via(wordCountFlow).runWith(Sink.head)
  val g7 = wordCountFlow.runWith(sentenceSource, Sink.head)._2

  system.terminate()
}